package com.ksc.wordcount.shuffle.nettyimpl.client;

import com.ksc.wordcount.shuffle.ShuffleBlockId;
import com.ksc.wordcount.task.KeyValue;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

import java.util.AbstractMap;
import java.util.stream.Stream;

public class ShuffleClient {

    // 声明一个Bootstrap对象，用于创建和配置客户端的网络连接
    // static  Bootstrap clientBootstrap = getNettyBootStrap();

    // 获取Netty Bootstrap对象
    public  Bootstrap getNettyBootStrap(){
        // 创建一个EventLoopGroup实例，用于处理I/O操作的多线程事件循环
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(group);
        return b;
    }

    // 根据给定的ShuffleBlockId，从远程服务器获取对应的Shuffle数据流
    public Stream<KeyValue> fetchShuffleData(ShuffleBlockId shuffleBlockId) throws InterruptedException {
        // 创建ShuffleClientHandler实例，用于处理客户端与服务器之间的网络通信
        ShuffleClientHandler shuffleClientHandler = new ShuffleClientHandler();

        // 获取Netty Bootstrap对象，并配置连接到远程服务器的相关信息和处理器
        Bootstrap bootstrapChannel = getNettyBootStrap().channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        // 添加ObjectEncoder和ObjectDecoder用于序列化和反序列化对象
                        ch.pipeline().addLast(
                                new ObjectEncoder(),
                                new ObjectDecoder(ClassResolvers.cacheDisabled(null)),
                                shuffleClientHandler
                        );
                    }
                });

        // 连接到远程服务器
        ChannelFuture channelFuture = bootstrapChannel.connect(shuffleBlockId.getHost(), shuffleBlockId.getPort()).sync();

        // 监听连接结果
        channelFuture.addListener(future -> {
            if (future.isSuccess()) {
                System.out.println("connect File Server: 连接服务器成功");
            } else {
                System.out.println("connect File Server: 连接服务器失败");
            }
        });

        // 发送ShuffleBlockId对象到服务器
        channelFuture.channel().writeAndFlush(shuffleBlockId);
        System.out.println("connect File Server: 已发送文件请求");

        // 关闭连接
        // channelFuture.channel().closeFuture().sync();
        channelFuture.channel().closeFuture();

        // 返回从服务器获取的Shuffle数据流
        return shuffleClientHandler.getStream();
    }
}